/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */

package cgl.narada.transport.ssl;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.security.Security;
import java.util.Properties;

import javax.net.ssl.SSLSocketFactory;

import org.apache.log4j.Logger;

import cgl.narada.transport.TransportException;
import cgl.narada.transport.proxy.ProxyTunnel;
import cgl.narada.transport.sslHttpBase.LinkMessageKeepAlive;
import cgl.narada.transport.sslHttpBase.Transport;
import cgl.narada.transport.sslHttpBase.TransportDataListener;
import cgl.narada.transport.sslHttpBase.TransportImpl;
import cgl.narada.transport.util.BytesUtil;
import cgl.narada.util.SystemInit;
import cgl.narada.util.SystemNotInitializedException;

/**
 * @author John Yin, Hongbin Liu
 * @author Harshawardhan Gadgil $Revision$
 */

public class SSLTransportJSSEImpl extends TransportImpl implements SSLTransport {

	final static Logger log = Logger.getLogger(SSLTransportJSSEImpl.class
			.getName());

	private ReceiverThread m_receiverThread = new ReceiverThread("sslreceiver");
	private SenderThread m_senderThread = new SenderThread("sslsender");
	private DeliveryThread m_deliveryThread = null;
	private TransportDataListener m_listener = null;

	// Beyond this size means the message is corrupt.
	private static final int MAX_PAYLOAD_SIZE = 10000000;
	private Socket m_socket;
    private int socketTimeout = 4000;
	/**
	 * -------------------------------------------------------------------
	 */
	public SSLTransportJSSEImpl() {}

	/**
	 * ------------------------------------------------------------------- Used
	 * when an SSL transport end point is accepted rather than initiated.
	 */
	protected SSLTransportJSSEImpl(Socket socket) {
		m_socket = socket;
        
        try {
            socket.setSoTimeout(socketTimeout);
        } catch (SocketException e1) {
            System.out.println("SSLTransportJSSEImpl: Problems setting timeout");
        }
        
		try {
			startThreads();
		}
		catch (IOException e) {
			log.error("Unable to start send/receive thread");
		}
	}

	public void setProperties(Properties properties) {
		super.setProperties(properties);
		// Properties systemProperties = System.getProperties();

		String val = properties.getProperty("initiator.trustStore");
		if (val != null && val.length() > 0) {
			System.setProperty("javax.net.ssl.trustStore", val);
			log.debug("Setting trustStore to " + val);
		}

		val = properties.getProperty("initiator.trustStorePassword");
		if (val != null && val.length() > 0) {
			System.setProperty("javax.net.ssl.trustStorePassword", val);
			log.debug("Setting trustStore password to " + val);
		}

		if (System.getProperty("javax.net.ssl.trustStore") == null) {

			if (properties.containsKey("controlserver.port")) {
				log.info("Control Server Port not specified. "
						+ "Will use default broker Keystore/Trustore");

				String trustStore = null;
				String password = null;
				try {
					trustStore = SystemInit
							.getInitData(SystemInit.SYS_BROKER_CERTIFICATE_TRUSTRORE);
					password = SystemInit
							.getInitData(SystemInit.SYS_BROKER_KEYSTORE_PASSWORD);
				}
				catch (SystemNotInitializedException e) {
					e.printStackTrace();
				}

				log.debug("javax.net.ssl.trustStore set to " + trustStore);
				System.setProperty("javax.net.ssl.trustStore", trustStore);

				log
						.debug("javax.net.ssl.trustStorePassword set to "
								+ password);
				System
						.setProperty("javax.net.ssl.trustStorePassword",
								password);

			}
			else {

				int port = Integer.parseInt(properties
						.getProperty("controlserver.port"));

				String certFile = getCert(m_host, port);

				if (certFile != null) {
					log.debug("javax.net.ssl.trustStore set to " + certFile);
					System.setProperty("javax.net.ssl.trustStore", certFile);
					String trustStorePass = SSLJSSEControl.getInstance()
							.trustStorePass();
					log.debug("javax.net.ssl.trustStorePassword set to "
							+ trustStorePass);
					System.setProperty("javax.net.ssl.trustStorePassword",
							trustStorePass);
				}
			}
		}

		// HG: Modified
		// Register JSSE
		Security.addProvider(new com.sun.net.ssl.internal.ssl.Provider());

		// Simply set the protocol handler property to use SSL.
		System.setProperty("java.protocol.handler.pkgs",
				"com.sun.net.ssl.internal.www.protocol");

		m_sslFactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
		log.debug("Setting to secure mode");
	}

	private String getCert(String host, int port) {
		String certFile = null;
		try {
			certFile = new ControlClient().getCertificate(host, port);
		}
		catch (Exception e) {
			log.warn("having problem with obtaining certificate from " + host);
			e.printStackTrace();
		}
		return certFile;
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public void startThreads() throws IOException {
		m_receiverThread.start();
		m_senderThread.start();
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public void stopThreads() throws IOException {
		m_receiverThread.destroy();
		m_senderThread.destroy();
		if (m_deliveryThread != null) m_deliveryThread.destroy();
		if (m_listener != null) m_listener.transportClosed(this);
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public void setTransportDataListener(TransportDataListener l) {
		if (m_listener == null && l != null) {
			m_deliveryThread = new DeliveryThread(this);
			m_deliveryThread.start();
		}
		else if (l == null && m_deliveryThread != null) {
			m_deliveryThread.destroy();
			m_deliveryThread = null;
		}
		m_listener = l;
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public boolean isConnected() {
		return m_socket != null;
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public boolean isInErrorState() {
		return m_ioerror;
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public String getType() {
		return SSLLinkFactory.CHANNEL_TYPE;
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public void connect() throws TransportException {
		if (m_host == null) {
			throw new TransportException("Unspecified endpoint host");
		}
		connect(m_host, m_port);
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public void connect(String host, int port) throws TransportException {
		/*
		 * Set up a socket to do tunneling through the proxy. Start it off as a
		 * regular socket, then layer SSL over the top of it.
		 */
		String proxyHost = proxyInfo.getProxyHost();
		try {
			if (proxyHost != null) {
				proxyTunnel = new ProxyTunnel(proxyInfo);
				m_socket = proxyTunnel.doProxyConnect(host, port);
				log.debug("Connected to proxy " + proxyHost + ":"
						+ proxyInfo.getProxyPort());
			}
			else {
				m_socket = new Socket(host, port);
                m_socket.setSoTimeout(socketTimeout);
				log.debug("Connected to server " + host + ":" + port);
			}

			m_socket.setSoLinger(false, 0);
			m_socket.setTcpNoDelay(true);

			if (m_sslFactory != null) {
				m_socket = setupSSL(m_socket, host, port);
                m_socket.setSoTimeout(socketTimeout);
				log.debug("Completed SSL handshake to " + host + ":" + port);
			}
			startThreads();

		}
		catch (IOException e) {
			e.printStackTrace();
			throw new TransportException("IOError: " + e.getMessage());
		}
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public void connect(String host) throws TransportException {
		connect(host, m_port);
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public synchronized void disconnect() throws IOException {
		if (m_socket != null) {
			stopThreads();
			m_socket.close();
			m_socket = null;
		}
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public boolean isSecure() {
		return (m_sslFactory != null);
	}

	/**
	 * -------------------------------------------------------------------
	 */
	public String toString() {
		if (m_host != null) {
			return "=>" + m_host + ":" + m_port;
		}
		else {
			return "<=" + m_socket.getInetAddress().getHostName() + ":"
					+ m_socket.getLocalPort();
		}
	}

	/**
	 * -------------------------------------------------------------------
	 */
	protected void throwBadConnectionException() throws IOException {
		if (m_ioerror) {
			throw new IOException("Connection in error state.  disconnected");
		}
		else {
			throw new IOException("Connection closed");
		}
	}

	/**
	 * -------------------------------------------------------------------
	 */
	protected void sendDataImmediately(byte[] message) throws IOException {
		if (m_socket == null) throwBadConnectionException();
		OutputStream out = m_socket.getOutputStream();
		out.write(BytesUtil.intToBytes(message.length));
		out.write(message);
		out.flush();
	}

	/**
	 * -------------------------------------------------------------------
	 */
	protected byte[] receiveDataImmediately() throws IOException {
		if (m_socket == null) throwBadConnectionException();
		InputStream in = m_socket.getInputStream();
		int size = BytesUtil.bytesToInt(readBytes(in, 4));
		if (size > MAX_PAYLOAD_SIZE)
			throw new IOException("Bad size or payload too large: " + size);
		return readBytes(in, size);
	}

	/**
	 * -------------------------------------------------------------------
	 */
	protected void handleReceiveException(Exception e) {
		// Handle by closing the connection.
		try {
			m_ioerror = true;
			disconnect();
		}
		catch (IOException ex) {}
	}

	/**
	 * -------------------------------------------------------------------
	 */
	protected void handleSendException(Exception e) {
		// Handle by closing the connection.
		try {
			m_ioerror = true;
			disconnect();
		}
		catch (IOException ex) {}
	}

	/**
	 * --------------------------------------------------------------------
	 * Reads a fix number of bytes from an input stream. Will block if the
	 * amount can not be fully read.
	 */
	private byte[] readBytes(InputStream input, int size) throws IOException {
		byte[] buf = new byte[size];

		int remainingSize = size;
		int readSize = 0;
		while ((readSize = input.read(buf, (size - remainingSize),
				remainingSize)) < remainingSize) {
			if (readSize < 0) {
				throw new IOException(
						"Stream terminated before chunk can be fully read[size="
								+ size + "]");
			}
			remainingSize -= readSize;
		}
		return buf;
	}

	/**
	 * =======================================================================
	 * Helper thread to take data off of the queue and send it.
	 */
	class SenderThread extends Thread {
		private boolean m_stopped = false;
		private long m_lastStatusCheck = System.currentTimeMillis();

		public SenderThread(String name) {
			super(name);
		}

		public void run() {
			while (!m_stopped) {
				byte[] data = m_sendQueue.removeData(1000);

				long currentTime = System.currentTimeMillis();
				try {
					if (m_statusInterval != 0
							&& currentTime - m_lastStatusCheck > m_statusInterval) {
						statusCheck();
						m_lastStatusCheck = currentTime;
					}

					if (data == null) continue;
					sendDataImmediately(data);
				}
				catch (IOException e) {
					e.printStackTrace();
					log.error("Sending exception occurred " + e.getMessage());
					handleSendException(e);
				}
			}
		}

		/**
		 * -------------------------------------------------------------------
		 */
		public void statusCheck() throws IOException {
			// Send a keep alive message.
			LinkMessageKeepAlive kam = new LinkMessageKeepAlive();
			byte[] bytes = kam.toBytes();
			sendDataImmediately(kam.toBytes());
		}

		/**
		 * -------------------------------------------------------------------
		 */
		public void destroy() {
			if (!m_stopped) {
				m_stopped = true;
				interrupt();
			}
		}
	}

	/**
	 * =======================================================================
	 * Helper thread to take data off of the queue and send it.
	 */
	class ReceiverThread extends Thread {
		private boolean m_stopped = false;

		public ReceiverThread(String name) {
			super(name);
		}

		public void run() {
			while (!m_stopped) {
				try {
					byte[] data = receiveDataImmediately();
					// If it is a transport level system message then process
					// and ignore it.
					if (!handleTransportSystemMessage(data)) {
						m_receiveQueue.addData(data);
					}
				}
				catch (IOException e) {
					log.warn("Read error: " + e.getMessage());
					handleReceiveException(e);
				}
			}
		}

		/**
		 * -------------------------------------------------------------------
		 */
		public void destroy() {
			if (!m_stopped) {
				m_stopped = true;
				interrupt();
			}
		}
	}

	/**
	 * =======================================================================
	 * Helper thread to take data off of the receive queue and invoke data
	 * received on the listener.
	 */
	class DeliveryThread extends Thread {
		private boolean m_stopped = false;
		private Transport m_parent = null;

		/**
		 * -------------------------------------------------------------------
		 */
		DeliveryThread(Transport parent) {
			m_parent = parent;
			setName("ssldelivery");
		}

		/**
		 * -------------------------------------------------------------------
		 */
		public void run() {
			while (!m_stopped) {
				byte[] data = m_receiveQueue.removeData();
				if (data != null) {
					m_listener.dataReceived(m_parent, data);
				}
				else {
					if (!m_stopped) {
						log
								.warn("Delivery thread received null data.  Exiting");
					}
					destroy();
				}
			}
		}

		/**
		 * -------------------------------------------------------------------
		 */
		public void destroy() {
			if (!m_stopped) {
				m_stopped = true;
				m_receiveQueue.dispose();
			}
		}
	}

}
